package liquidfs.core;

import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;

import liquidfs.*;
import liquidfs.data.*;
import liquidfs.data.ex.*;
import liquidfs.data.ops.*;
import liquidfs.ui.net.*;
import utility.*;

public class OperationProcessor implements Runnable, ParameterizedRunnable<MessageClient> {
	
   private static final int queueCapacity = 1024;
	private static Logger log = Logger.getLogger("liquidfs.core.OperationProcessor");
	private Application app;
	private ConcurrentMap<PeerDevice, OperationQueue> queues = new ConcurrentHashMap<PeerDevice, OperationQueue>();
	private List<ModificationListener> listeners = new ArrayList<ModificationListener>(10);
	
	public OperationProcessor (Application app) {
		this.app = app;
		
		// Search for any pre-existing queues in the filesystem
		File queuesDirectory = new File(app.getConfiguration().getQueuePath());
		
		// If the directory doesn't exist, create it
		if (!queuesDirectory.exists())
		   queuesDirectory.mkdir();
		
		for (File file : queuesDirectory.listFiles()) {
		   
		   // Skip things that aren't queues
		   if (file.isDirectory() || !file.getName().endsWith("_queue.xmls"))
		      continue;
		   
		   // Parse the ID out of the file name
		   String deviceID = file.getName().replaceFirst("_queue.xmls", "");
		   
		   // A link containing this device ID to be searched for
		   Link deviceLink = null;
		   
		   // Check if we are linked to a device with that ID
		   for (Link link : app.getConfiguration().getLinks()) {
		      if (link.getDevice().getID().equalsIgnoreCase(deviceID)) {
		         deviceLink = link;
		         break;
		      }
		   }
		   
		   // Attempt to delete this queue if we are not linked to this device
		   if (deviceLink == null) {
		      file.delete();
		      continue;
		   }
		   
		   // If we have found a link, then load the queue into memory and add it to our map of queues
		   try {
		      OperationQueue queue = OperationQueue.loadQperationQueue(app, deviceLink.getDevice(), queueCapacity, file.getAbsolutePath());
		      queues.put(deviceLink.getDevice(), queue);
		   }
		   catch (IOException e) {
		      throw new UserUncheckedException("Unable to load queue from journal: " + e.getMessage(), e);
		   }
		}
	}
	
	public void addModificationListener (ModificationListener listener) {
	   this.listeners.add(listener);
	}
	
	public void removeModificationListener (ModificationListener listener) {
	   this.listeners.remove(listener);
	}
	
	public void fireObjectModified (String path) {
	   for (ModificationListener listener : listeners) {
	      listener.objectModified(path);
	   }
	}
	
	public <T> Result<T> process (Operation<T> operation, InputStream in, OutputStream out) {
		try {
		   
		   // If we've been reached before, we can ignore this request
	      if (operation.reached(app.getConfiguration().getLocalDevice())) {
	         return new ExceptionResult<T>(operation, new AlreadyRespondedException());
	      }
	      
	      // Add this device to the list of devices that have been reached so far, so that it doesn't get queried again by other devices on the network
	      operation.addReachedDevice(app.getConfiguration().getLocalDevice());
	      
	      // If the device is a modification, we need to check for conflicts and to make sure we haven't received this same message before
	      if (operation.isModification()) {
	         
	         // Cast it to a modification operation
	         ModificationOperation<T> mod = (ModificationOperation<T>) operation;
	         
	         // Check if this is conflicting
	         if (mod.isConflicting(this.app)) {
	            
	            // If so, let the sender know
	            return new ExceptionResult<T>(mod, new ConflictException());
	         }
	         
	         // Check if it is not a direct conflict but must be ignored for some other reason (it is old, from a conflicting branch, etc.)
	         if (mod.isIgnored(this.app)) {
	            
	            // If so, let the sender know we ignored it
	            return new ExceptionResult<T>(mod, new IgnoredException());
	         }
	      }
	      
	      // Execute the operation
		   T object = operation.execute (this.app, in, out);
		   
		   // Notify anyone interested in file modifications
		   if (operation.isModification() && operation.isFileOperation()) {
		      FileOperation<?> fop = (FileOperation<?>) operation;
		      fireObjectModified(fop.getFilePath());
		   }
		   
		   // Create the result out of the object that the operation returned
		   Result<T> result = new Result<T>(operation, object);
		   
		   // Return the result to the caller
		   return result;
		}
		catch (FileSystemException e)	{
		   return new ExceptionResult<T>(operation, e);
		}
	}
	
	public <T> Result<T> relay (Link link, Operation<T> op, InputStream in, OutputStream out) {
	   
	   // Don't do anything if that device has already been reached
	   if (op.reached(link.getDevice())) {
	      return new ExceptionResult<T>(op, new AlreadyRespondedException());
	   }
	   
	   // Mark this device as having been reached, if it hasn't been marked as such already (i.e., if it wasn't processed, only relayed)
	   op.addReachedDevice(this.app.getConfiguration().getLocalDevice());
	   
	   // If it is a read operation, simply send it if possible and return a result or an error
	   if (!op.isModification()) {
	      try {
	         return send(link, op, in, out);
	      }
	      catch (CommunicationException e) {
	         return new ExceptionResult<T>(op, e);
	      }
	   }
	   
	   // For write operations, we need to be careful about them being received in order, so they are queued
	   OperationQueue queue = queues.get(link.getDevice());
	   
	   // If there is nothing in the queue (or we haven't even needed to create a queue yet), we can simply send it right away without worry
	   if (queue == null || queue.isEmpty()) {
	      try {
	         return send(link, op, in, out);
	      }
	      // If it fails, add it to the queue and let the caller know that they're waiting
	      catch (CommunicationException e) {
	         
	         // If there isn't a queue, create one
	         if (queue == null) {
   	         queue = new OperationQueue(this.app, link.getDevice(), queueCapacity);
   	         queues.put(link.getDevice(), queue);
   	      }
	         
	         queue.add(new BufferedOperation(op, in));
	         return new WaitingResult<T>(op);
	      }
	   }
	   
	   // If the queue is not empty, let the caller know that they are waiting, other things need to be send first
	   // (and chances are, the host is down anyway, which is why the queue isn't empty)
	   queue.add(new BufferedOperation(op, in));
	   return new WaitingResult<T>(op);
	}
	
	protected <T> Result<T> send (Link link, Operation<T> op, InputStream in, OutputStream out) throws CommunicationException {
	   MessageClient client;
      
	   // Check if the host is worth trying (i.e., did it fail recently?)
	   if (!link.getHost().isWorthTrying())
	      throw new HostDownException("The device " + link.toString() + " was not tried again so soon because it was recently down.");
	   
	   // Connect to the host
      try {
         client = new MessageClient(app, link.getHost());
      }
      catch (IOException e) {
         link.getHost().setLastCommunication(false);
         throw new HostDownException("The device " + link.toString() + " appears to be down.", e);
      }
      
      // Send the message
      try {
         client.send(new OperationMessage(op), in);
      }
      catch (IOException e) {
         link.getHost().setLastCommunication(false);
         throw new CommunicationException("Could not send to device " + link.toString() + "; it appears to be down.", e);
      }
      
      // Get the result
      ResultMessage message;
      
      try {
         message = (ResultMessage) client.receive(out);
      } 
      catch (IOException e) {
         link.getHost().setLastCommunication(false);
         throw new CommunicationException("The device " + link.toString() + " did not respond properly.", e);
      }
      
      // Close the connection
      try {
         client.close();
      }
      catch (IOException e) {
         link.getHost().setLastCommunication(false);
         throw new CommunicationException("The device " + link.toString() + " could not be closed properly.", e);
      }
      
      // Mark this as a successful exchange with the host
      link.getHost().setLastCommunication(true);
      
      @SuppressWarnings("unchecked")
      Result<T> result = (Result<T>) message.getResult();
      
      // Add the devices that the operation reached back to the original operation
      op.addReachedDevices(result.getReachedDevices());
      
      // Return the result
      return result;
	}
	
	@Override
	public void run () {
	   
	   // Loop forever
	   while (true) {
	      
	      // Keep track of whether anything is unempty
	      boolean wait = true;
	      
	      // Get all of the hosts that we have queues for
	      Set<Map.Entry<PeerDevice, OperationQueue>> entries = queues.entrySet();
	      
	      // Iterate through each of them
	      for (Map.Entry<PeerDevice, OperationQueue> entry : entries) {
	         
	         // Get the queue for this device
            OperationQueue queue = entry.getValue();
            
            // Get the next operation in the queue
            BufferedOperation op = queue.peek();
            
            // If there's nothing in it, move on to the next one
            if (op == null)
               continue;
	         
	         for (Link link : this.app.getConfiguration().getLinks())
	         {
	            if (!link.getDevice().equals(queue.getDevice()))
	               continue;
   	         
   	         // Try to relay the operation
   	         try {
   	            this.send(link, op.getInternalOperation(), op.getInputStream(), null);
   	         }
   	         // If it fails again, just move on to the next queue
   	         catch (CommunicationException e) {
   	            continue;
   	         }
   	         
   	         // If it succeeded, remove that operation from the queue
   	         queue.remove();
   	         
   	         // Check if there's still something in the queue
   	         if (!queue.isEmpty())
   	            wait = false;
   	         
   	         // If we've successfully delivered it, we don't need to check the other links for this same device
   	         break;
	         }
	      }
	      
	      // Sleep if there's nothing to do
	      if (wait) {
            try {
               Thread.sleep(20);
            }
            catch (InterruptedException e) {
               // Do nothing
            }
	      }
	   }
	   
	}
	
	public List<OperationQueue> getQueues () {
	   
	   List<OperationQueue> list = new ArrayList<OperationQueue>(this.queues.size());
	   
	   for (OperationQueue queue : this.queues.values()) {
	      list.add(queue);
	   }
	   
	   return list;
	   
	}
	
	@Override
	public void run(MessageClient client) {
		OperationMessage message;
		
		// Read a message from the client
		try {
			message = (OperationMessage) client.receive();
		}
		catch (IOException e) {
			log.log(Level.WARNING, e.getLocalizedMessage());
			return;
		}
		
		// Process it and obtain a result
		Result<?> result = this.process (message.getOperation(), client.getBinaryInput(), client.getBinaryOutput());
		
		// Send it
		try {
		   client.send(new ResultMessage(result));
		}
		catch (IOException e) {
         log.log(Level.WARNING, e.getLocalizedMessage());
         return;
      }
		finally {
		   try {
		      client.close();
		   }
		   catch (IOException e) {
	         log.log(Level.WARNING, e.getLocalizedMessage());
	         return;
	      }
		}
	}

}
